﻿using Confluent.Kafka.Admin;
using Confluent.Kafka;
using LJ.Abstractions.Config;
using LJ.EventBus.Data;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using LJ.EventBus.HornedSerpent.Transport.Abstractions;

namespace LJ.EventBus.HornedSerpent.Transport.Kafka.Internal;

public class KafkaEventBusConsumerClient(IConfigProvider configProvider) : IEventBusConsumerClient
{
    private static readonly object Lock = new();
    private string _group = null!;
    private string _topic = null!;
    private readonly EventBusTransportConfig _eventBusTransportConfig = configProvider.Get<EventBusTransportConfig>();

    private IConsumer<string, byte[]>? _consumerClient;
    private bool disposedValue;

    internal void Init(string group, string topic)
    {
        _group = group;
        _topic = topic;
    }

    public Func<EventBusTransportData, object, Task>? OnDataCallback { get; set; }
    public Action<EventBusTransportLog>? OnLogCallback { get; set; }

    public void FetchTopic()
    {

        try
        {
            var config = new AdminClientConfig(_eventBusTransportConfig.MainConfig) { BootstrapServers = _eventBusTransportConfig.Servers };

            using var adminClient = new AdminClientBuilder(config).Build();

            adminClient.CreateTopicsAsync([new TopicSpecification
            {
                Name = _topic,
                NumPartitions = _eventBusTransportConfig.Topic.NumPartitions,
                ReplicationFactor = _eventBusTransportConfig.Topic.ReplicationFactor
            }]).GetAwaiter().GetResult();
        }
        catch (CreateTopicsException ex) when (ex.Message.Contains("already exists"))
        {
        }
        catch (Exception ex)
        {
            var log = new EventBusTransportLog(
                "An error was encountered when automatically creating topic! -->" + ex.Message,
                EventBusTransportLogType.ConsumeError);

            OnLogCallback!(log);
        }
    }
    public Task ListeningAsync(TimeSpan timeout, CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        Connect();

        while (!cancellationToken.IsCancellationRequested)
        {
            ConsumeResult<string, byte[]> consumerResult;

            try
            {
                consumerResult = _consumerClient!.Consume(timeout);

                if (consumerResult == null) continue;
                if (consumerResult.IsPartitionEOF || consumerResult.Message.Value == null) continue;
            }
            catch (ConsumeException e) when (_eventBusTransportConfig.RetriableErrorCodes.Contains(e.Error.Code))
            {
                var log = new EventBusTransportLog(
                    e.Error.ToString(),
                    EventBusTransportLogType.ConsumeRetries);

                OnLogCallback!(log);

                continue;
            }

            Consume(consumerResult).GetAwaiter().GetResult();
        }
        return Task.CompletedTask;
    }
   

    public void Subscribe()
    {

        Connect();

        _consumerClient!.Subscribe([_topic]);
    }


    public void Reject(object? sender)
    {
        _consumerClient!.Assign(_consumerClient.Assignment);
    }

    public void Commit(object? sender)
    {
        _consumerClient!.Commit((ConsumeResult<string, byte[]>)sender!);
    }

    private void Connect()
    {
        if (_consumerClient != null) return;

        lock (Lock)
        {
            if (_consumerClient == null)
            {
                var config = new ConsumerConfig(new Dictionary<string, string>(_eventBusTransportConfig.MainConfig));
                config.BootstrapServers ??= _eventBusTransportConfig.Servers;
                config.GroupId ??= _group;
                config.AutoOffsetReset ??= AutoOffsetReset.Earliest;
                config.AllowAutoCreateTopics ??= true;
                config.EnableAutoCommit ??= false;
                config.LogConnectionClose ??= false;

                _consumerClient = new ConsumerBuilder<string, byte[]>(config)
                .SetErrorHandler(ConsumerClient_OnConsumeError)
                .Build();
            }
        }
    }

    private void ConsumerClient_OnConsumeError(IConsumer<string, byte[]> consumer, Error e)
    {
        var log = new EventBusTransportLog(
            $"An error occurred during connect kafka --> {e.Reason}",
            EventBusTransportLogType.ServerConnError);
        OnLogCallback!(log);
    }


    private async Task Consume(ConsumeResult<string, byte[]> consumerResult)
    {
        var headers = new Dictionary<string, string?>(consumerResult.Message.Headers.Count);
        foreach (var header in consumerResult.Message.Headers)
        {
            var val = header.GetValueBytes();
            headers[header.Key] = val != null ? Encoding.UTF8.GetString(val) : null;
        }

        if (!headers.ContainsKey(EventBusHeaders.MessageName))
        {
            headers[EventBusHeaders.MessageName] = _topic;
        }


        headers[EventBusHeaders.Group] = _group;

        var data = new EventBusTransportData(headers, consumerResult.Message.Value);

        await OnDataCallback!(data, consumerResult);
    }

    #region Dispose
    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
                _consumerClient?.Dispose();
            }

            // TODO: 释放未托管的资源(未托管的对象)并重写终结器
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
    // ~KafkaEventBusConsumerClient()
    // {
    //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
    //     Dispose(disposing: false);
    // }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion

}
